package org.chen.task;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.chen.client.HeyGenClient;
import org.chen.common.constant.VideoStatusEnum;
import org.chen.domain.entity.VideoArticle;
import org.chen.domain.entity.VideoTask;
import org.chen.mapper.ArticleMapper;
import org.chen.mapper.VideoTaskMapper;
import org.chen.response.VideoStatusResponse;
import org.chen.utils.VideoRedisUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PreDestroy;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.*;

@Component
@Slf4j
public class VideoStatusCheckTask {
    @Autowired
    private VideoRedisUtils videoRedisUtils;
    
    @Autowired
    private HeyGenClient heyGenClient;
    
    @Autowired
    private VideoTaskMapper taskMapper;

    @Autowired
    private ArticleMapper articleMapper;

    // 可注入配置属性
    @Value("${video.check.timeout-hours:2}")
    private int taskTimeoutHours;

    @Value("${video.check.batch-size:50}")
    private int batchSize;

    private final ExecutorService executorService;

    public VideoStatusCheckTask() {
        this.executorService = Executors.newFixedThreadPool(5);
    }

    @Scheduled(fixedRate = 300000) // 每5分钟执行一次
    @Transactional(rollbackFor = Exception.class)
    public void checkVideoStatus() {
        Set<String> pendingVideoIds = videoRedisUtils.getAllPendingVideoIds();
        log.info("Starting to check video status for {} videos", pendingVideoIds.size());

        // 分批处理，避免一次处理太多
        List<String> videoIdList = new ArrayList<>(pendingVideoIds);
        // 使用计数器来等待所有任务完成
        CountDownLatch latch = new CountDownLatch(videoIdList.size());

        // 分批处理任务和批量更新结果
        List<VideoTask> tasksToUpdate = Collections.synchronizedList(new ArrayList<>());
        Map<String, Integer> articleStatusUpdates = new ConcurrentHashMap<>();

        for (String videoId : videoIdList) {
            executorService.submit(() -> {
                try {
                    VideoTask task = taskMapper.selectOne(
                            new LambdaQueryWrapper<VideoTask>()
                                    .eq(VideoTask::getVideoId, videoId)
                    );

                    if (task == null) {
                        videoRedisUtils.removeVideoId(videoId);
                        return;
                    }

                    if (task.getStatus().equals(VideoStatusEnum.COMPLETED.getCode())) {
                        videoRedisUtils.removeVideoId(videoId);
                        return;
                    }

                    processVideoStatus(task, videoId);

                    // 添加到批量更新列表
                    if (task.getStatus().equals(VideoStatusEnum.COMPLETED.getCode()) ||
                            task.getStatus().equals(VideoStatusEnum.FAILED.getCode())) {
                        tasksToUpdate.add(task);
                        articleStatusUpdates.put(task.getArticleId(), task.getStatus());
                    }
                } catch (Exception e) {
                    log.error("Check video status failed for video_id: {}", videoId, e);
                    videoRedisUtils.removeVideoId(videoId);
                } finally {
                    latch.countDown();
                }
            });
        }

        try {
            // 等待所有任务完成，最多等待5分钟
            boolean completed = latch.await(5, TimeUnit.MINUTES);
            if (!completed) {
                log.warn("Not all video status checks completed within the timeout");
            }

            // 批量更新数据库
            if (!tasksToUpdate.isEmpty()) {
                batchUpdateTasks(tasksToUpdate, articleStatusUpdates);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("Interrupted while waiting for video status checks to complete", e);
        }
    }

    @Scheduled(fixedRate = 300000) // 每5分钟执行一次
    public void checkPendingTasks() {
        log.info("Starting to check pending tasks...");

        // 修正查询条件 - 应该是查询生成中的任务
        List<VideoTask> pendingTasks = taskMapper.selectList(new LambdaQueryWrapper<VideoTask>()
                .eq(VideoTask::getStatus, VideoStatusEnum.GENERATING.getCode()));

        log.info("Found {} pending tasks", pendingTasks.size());

        if (pendingTasks.isEmpty()) {
            return;
        }

        // 分批处理，每次处理batchSize个任务
        List<List<VideoTask>> batches = new ArrayList<>();
        for (int i = 0; i < pendingTasks.size(); i += batchSize) {
            batches.add(pendingTasks.subList(i, Math.min(pendingTasks.size(), i + batchSize)));
        }

        for (List<VideoTask> batch : batches) {
            processPendingTasksBatch(batch);
        }
    }

    private void processPendingTasksBatch(List<VideoTask> tasks) {
        CountDownLatch batchLatch = new CountDownLatch(tasks.size());
        List<VideoTask> tasksToUpdate = Collections.synchronizedList(new ArrayList<>());
        Map<String, Integer> articleStatusUpdates = new ConcurrentHashMap<>();

        for (VideoTask task : tasks) {
            executorService.submit(() -> {
                try {
                    processVideoStatus(task, task.getVideoId());

                    // 添加到批量更新列表
                    if (task.getStatus().equals(VideoStatusEnum.COMPLETED.getCode()) ||
                            task.getStatus().equals(VideoStatusEnum.FAILED.getCode())) {
                        tasksToUpdate.add(task);
                        articleStatusUpdates.put(task.getArticleId(), task.getStatus());
                    }
                } catch (Exception e) {
                    log.error("Failed to check task status: {}", task.getId(), e);
                } finally {
                    batchLatch.countDown();
                }
            });
        }

        try {
            // 等待当前批次完成
            boolean await = batchLatch.await(2, TimeUnit.MINUTES);

            // 批量更新数据库
            if (!tasksToUpdate.isEmpty()) {
                batchUpdateTasks(tasksToUpdate, articleStatusUpdates);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("Interrupted while processing pending tasks batch", e);
        }
    }

    private void processVideoStatus(VideoTask task, String videoId) {
        try {
            VideoStatusResponse status = heyGenClient.getVideoStatus(task.getVideoId(), task.getApiKey());

            if (status == null || status.getData() == null) {
                return;
            }

            log.debug("Video status for {}: {}", videoId, status.getData().getStatus());

            switch (status.getData().getStatus()) {
                case "completed":
                    task.setStatus(VideoStatusEnum.COMPLETED.getCode());
                    task.setVideoUrl(status.getData().getVideo_url());
                    task.setGifUrl(status.getData().getGif_url());
                    task.setErrorMessage(null);
                    if (videoId != null) {
                        videoRedisUtils.removeVideoId(videoId);
                    }
                    break;

                case "failed":
                    task.setStatus(VideoStatusEnum.FAILED.getCode());
                    if (status.getData().getError() != null) {
                        task.setErrorMessage(status.getData().getError());
                    } else if (status.getData().getError_detail() != null) {
                        task.setErrorMessage(status.getData().getError_detail().getMessage());
                    }
                    if (videoId != null) {
                        videoRedisUtils.removeVideoId(videoId);
                    }
                    break;

                case "pending":
                case "processing":
                    // 如果任务创建时间超过配置的小时数仍在处理，标记为失败
                    if (task.getCreateTime().isBefore(LocalDateTime.now().minusHours(taskTimeoutHours))) {
                        task.setStatus(VideoStatusEnum.FAILED.getCode());
                        task.setErrorMessage("Task timeout after " + taskTimeoutHours + " hours");
                        if (videoId != null) {
                            videoRedisUtils.removeVideoId(videoId);
                        }
                    }
                    break;
            }
        } catch (Exception e) {
            log.error("Error checking status for task: {}", task.getId(), e);
            // 如果API调用失败且任务已经处理很久，标记为失败
            if (task.getCreateTime().isBefore(LocalDateTime.now().minusHours(taskTimeoutHours))) {
                task.setStatus(VideoStatusEnum.FAILED.getCode());
                task.setErrorMessage("API Error: " + e.getMessage());
                if (videoId != null) {
                    videoRedisUtils.removeVideoId(videoId);
                }
            }
        }
    }

    /**
     * 批量更新任务和文章状态
     */
    private void batchUpdateTasks(List<VideoTask> tasks, Map<String, Integer> articleUpdates) {
        if (tasks.isEmpty()) {
            return;
        }

        try {
            // 批量更新任务
            taskMapper.updateById(tasks);
            log.info("Updated {} tasks in batch", tasks.size());

            // 批量更新文章状态
            if (!articleUpdates.isEmpty()) {
                List<VideoArticle> articlesToUpdate = new ArrayList<>();

                // 批量查询文章
                List<String> articleIds = new ArrayList<>(articleUpdates.keySet());
                List<VideoArticle> articles = articleMapper.selectBatchIds(articleIds);

                // 更新状态
                for (VideoArticle article : articles) {
                    Integer newStatus = articleUpdates.get(article.getId());
                    if (newStatus != null) {
                        article.setStatus(newStatus);
                        articlesToUpdate.add(article);
                    }
                }

                // 批量更新
                if (!articlesToUpdate.isEmpty()) {
                    articleMapper.updateById(articlesToUpdate);
                    log.info("Updated {} articles in batch", articlesToUpdate.size());
                }
            }
        } catch (Exception e) {
            log.error("Failed to perform batch updates", e);
            // 如果批量更新失败，尝试单个更新
            for (VideoTask task : tasks) {
                try {
                    taskMapper.updateById(task);
                    articleMapper.updateStatus(task.getArticleId(), task.getStatus());
                } catch (Exception ex) {
                    log.error("Failed to update task individually: {}", task.getId(), ex);
                }
            }
        }
    }

    // 优雅关闭线程池
    @PreDestroy
    public void shutdown() {
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                executorService.shutdownNow();
            }
        } catch (InterruptedException e) {
            executorService.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}